package psn.zjk.common.disuptor;

import com.lmax.disruptor.EventTranslator;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import psn.zjk.common.mq.rabbitmq.D;

public class DisruptorDemo {
    public Disruptor disruptor;
    int x = 0;
    int z = 0;

    public static void main(String[] args) throws InterruptedException {
        new DisruptorDemo().x();
    }

    public void x() throws InterruptedException {
        //消费者
        WorkHandler<D> eventHandler = new WorkHandler<D>() {
            @Override
            public void onEvent(D s) throws Exception {
                System.out.println("recv:" + s.getStatus());
                Thread.sleep(30);
            }
        };
        //多生产者模式，等待策略：yield
        disruptor = new Disruptor<>(this::get, 2, this::thread, ProducerType.MULTI, new YieldingWaitStrategy());
        disruptor.handleEventsWithWorkerPool(eventHandler);
        disruptor.start();

        //生产者
        new Thread(() -> {
            while (true) {
                disruptor.publishEvent(new EventTranslator<D>() {
                    @Override
                    public void translateTo(D o, long l) {
                        if (disruptor.getRingBuffer().hasAvailableCapacity(1)) {
                            o.a("x" + (x++));
                            System.out.println("publishx:" + x);
                        }
                    }
                });
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(() -> {
            while (true) {
                disruptor.publishEvent(new EventTranslator<D>() {
                    @Override
                    public void translateTo(D o, long l) {
                        o.a("z" + (z++));
                        System.out.println("publishz:" + z);
                    }
                });
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }


    public D get() {
        return new D();
    }

    public Thread thread(Runnable get) {
        return new Thread(get);
    }

}